package jgcp.master;

import java.io.File;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import jgcp.common.Task;
import jgcp.common.TaskDescription;
import jgcp.common.TaskStatus;
import jgcp.common.TaskStore;
import jgcp.common.network.Connection;
import jgcp.common.network.FileReceiver;
import jgcp.common.network.FileSender;
import jgcp.config.Configuration;
import jgcp.master.management.TaskQueue;
import jgcp.master.management.Worker_Task;
import jgcp.master.network.WorkerList;
import jgcp.master.service.MasterService;
import jgcp.message.Message;
import jgcp.message.MessageHandler;
import jgcp.message.MessageType;
import jgcp.security.PasswordAuthenticator;

/**
 * @Date 28/05/2009
 * @author Jie Zhao (288654)
 * @version 1.0
 */
public class MasterMessageHandler extends MessageHandler {
	private static AtomicInteger AI = new AtomicInteger(1);

	public MasterMessageHandler(Connection c) {
		super(c);
	}

	public void processMessage(Message m) throws Exception {
		if (m.getType() == MessageType.LOGIN) {
			PasswordAuthenticator pa = PasswordAuthenticator.getInstance();
			String[] buffer = m.getMessageBody().split(" ");
			Message mm = new Message();
			if (buffer.length == 2 && pa.login(buffer[0], buffer[1])) {
				mm.setType(MessageType.LOGIN_SUCCESS);
				mm.setMessageBody("Connection Successful.");
			} else {
				mm.setType(MessageType.LOGIN_REJECT);
				mm.setMessageBody("Incorrect Username or Password!");
			}
			connection.writeData(mm.toByteArray());
			System.out.println("New User Connected to master.");
		} else if (m.getType() == MessageType.LISTWORKER) {
			List<Integer> wl = WorkerList.getInstance().getWorkerList();
			StringWriter sw = new StringWriter();
			PrintWriter pw = new PrintWriter(sw);
			pw.println("###### Worker List ######");
			if (wl.size() == 0) {
				pw.println("Currently no worker connected to master.");
			}
			for (Integer i : wl) {
				Connection con = WorkerList.getInstance().getConnection(i);
				pw.println("Worker ID:" + i + "  IP Address:"
						+ con.getAddress() + " Port:" + con.getPort());
			}
			pw.println("###### List Ends ######");
			Message msg = new Message();
			msg.setType(MessageType.LISTWORKER_RESPONSE);
			msg.setMessageBody(sw.toString());
			connection.writeData(msg.toByteArray());
		} else if (m.getType() == MessageType.SUBMITTASK) {
			int port = Integer.parseInt(m.getMessageBody());
			String tempName = Configuration.getInstance().getMasterTempfolder()
					+ "temp" + AI.getAndIncrement() + ".zip";
			FileReceiver<TaskDescription> receiver = new FileReceiver<TaskDescription>(
					connection.getAddress(), port, tempName);
			receiver.start();
			while (receiver.isAlive())
				; // waiting for receive.
			Task t = new Task();
			TaskDescription td = receiver.getObject();
			t.setDescription(td);
			t.setFile(tempName);
			TaskStore.getInstance().store(t);
			TaskQueue.getInstance().put(t);
			Message msg = new Message();
			msg.setType(MessageType.SUBMITTASK_ACK);
			msg.setMessageBody("Your task is accepted. Task ID ="
					+ t.getTaskid());
			connection.writeData(msg.toByteArray());
		} else if (m.getType() == MessageType.LISTTASKS) {
			Message msg = new Message();
			msg.setType(MessageType.LISTTASKS_RESPONSE);
			StringWriter sw = new StringWriter();
			PrintWriter pw = new PrintWriter(sw);
			pw.println("###### Task List ######");
			List<Integer> wl = WorkerList.getInstance().getWorkerList();
			if (wl.size() == 0) {
				pw
						.println("No workers avaliable. The current list of tasks are waiting.");
				List<Integer> ww = TaskStore.getInstance().getStoreList();
				for (Integer i : ww) {
					pw.println("Task ID="
							+ i
							+ " | Status="
							+ TaskStore.getInstance().retrieve(i).getStatus()
									.toString());
				}
			}
			for (Integer i : wl) {
				pw.println("Worker ID:" + i + " is running following tasks:");
				List<Integer> taskids = Worker_Task.getInstance().getTasks(i);
				for (Integer j : taskids) {
					pw.println("--Task ID="
							+ j
							+ " | Status="
							+ TaskStore.getInstance().retrieve(j).getStatus()
									.toString());
				}
			}
			pw.println("###### List Ends ######");
			msg.setMessageBody(sw.toString());
			connection.writeData(msg.toByteArray());
		} else if (m.getType() == MessageType.QUERYSTATUS) {
			int taskId = Integer.parseInt(m.getMessageBody());
			Task task = TaskStore.getInstance().retrieve(taskId);
			Message msg = new Message();
			msg.setType(MessageType.QUERYSTATUS_RESPONSE);
			if (task != null) {
				msg.setMessageBody("Task ID=" + task.getTaskid() + " Status:"
						+ task.getStatus().toString());
			} else {
				msg.setMessageBody("Unknown Task ID.");
			}
			connection.writeData(msg.toByteArray());
		} else if (m.getType() == MessageType.CANCELTASK) {
			int taskId = Integer.parseInt(m.getMessageBody());
			int workerId = Worker_Task.getInstance().findWorker(taskId);
			if (workerId != -1) {
				Message msg = new Message();
				msg.setType(MessageType.CANCELTASK);
				msg.setMessageBody("" + taskId);
				new MasterService().sendMessage(workerId, msg);
			}
		} else if (m.getType() == MessageType.CANCELTASK_ACK) {
			int taskId = Integer.parseInt(m.getMessageBody());
			Task task = TaskStore.getInstance().retrieve(taskId);
			if (task != null) {
				task.setStatus(TaskStatus.CANCELED);
			}
		} else if (m.getType() == MessageType.GETRESULT) {
			int taskId = Integer.parseInt(m.getMessageBody());
			Task task = TaskStore.getInstance().retrieve(taskId);
			Message msg = new Message();
			if (task == null || task.getStatus() != TaskStatus.DONE) {
				msg.setType(MessageType.GETRESULT_NOTREADY);
				msg
						.setMessageBody("The task you need the result hasn't been finished. Please try again later.");
				connection.writeData(msg.toByteArray());
			} else {
				String resultFile = Configuration.getInstance().getMasterTempfolder()+"result_"+taskId;
				if(new File(resultFile).exists())
				{
					FileSender<String> fs = new FileSender<String>("",resultFile);
					fs.start();
					msg.setType(MessageType.GETRESULT_READY);
					msg.setMessageBody(""+fs.getPort());
					connection.writeData(msg.toByteArray());
				}
			}
		} else if (m.getType() == MessageType.SENDRESULT) {
			try {
				String[] buffer = m.getMessageBody().split(" ");
				int taskId = Integer.parseInt(buffer[0]);
				int port = Integer.parseInt(buffer[1]);
				String sourceAddress = connection.getSocket().getInetAddress().getHostAddress();
				String resultFile = Configuration.getInstance().getMasterTempfolder()+"result_"+taskId;
				FileReceiver<String> fr = new FileReceiver<String>(sourceAddress,port,resultFile);
				fr.start();
			} catch (Exception e) {
			}
		} else {
			int taskId = -1;
			Task task = null;
			switch (m.getType()) {
			case MessageType.STATUS_RUNNING:
				taskId = Integer.parseInt(m.getMessageBody());
				task = TaskStore.getInstance().retrieve(taskId);
				if (task != null)
					task.setStatus(TaskStatus.RUNNING);
				break;
			case MessageType.STATUS_ERROR:
				taskId = Integer.parseInt(m.getMessageBody());
				task = TaskStore.getInstance().retrieve(taskId);
				if (task != null)
					task.setStatus(TaskStatus.ERROR);
				break;
			case MessageType.STATUS_DONE:
				taskId = Integer.parseInt(m.getMessageBody());
				task = TaskStore.getInstance().retrieve(taskId);
				if (task != null)
					task.setStatus(TaskStatus.DONE);
				break;
			default:
				break;
			}
		}
	}
}
